package com.hkctp.producer;



import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class CustomerProducer {

    public static void main(String[] args){

        //配置信息
        Properties props = new Properties();
        //kafka集群
        props.put("bootstrap.servers","localhost:9092");
        //应答级别
        props.put(ProducerConfig.ACKS_CONFIG,"all");
        //重试次数-发送失败了会重试
        props.put("retries",0);
        /*
            一次发送的缓存
         */
        //批量大小
        props.put("batch.size",16384);
        //提交延时
        props.put("linger.ms",1);
        /*
            整个producer缓存多少数据
         */
        props.put("buffer.memory",33554432);
        //提供序列化的类   kv的序列化类
        props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");

        //创建生产者对象
        KafkaProducer<String,String> producer = new KafkaProducer<String,String>(props);

        //循环发送数据
        for (int i=0;i<10;i++){
            producer.send(new ProducerRecord<String, String>("second", String.valueOf(i)), new Callback() {
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if(e == null){
                        System.out.println(recordMetadata.partition()+"--"+recordMetadata.offset());
                    }else{
                       System.out.println("发送失败");
                    }
                }
            });
        }

        //关闭资源
        producer.close();

    }
}
